package cn.itcast.mr.combiner;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;

/**
 * Author itcast
 * Date 2021/10/17 15:02
 * Desc TODO
 */
public class BookMain {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //创建 Job 实例
        Job job = Job.getInstance();
        //job设置输入的格式化类型
        job.setInputFormatClass(TextInputFormat.class);
        //设置输入的文件路径
        TextInputFormat.addInputPath(job, new Path("D:\\project\\mr\\data\\books.txt"));
        //job设置map的实现类
        job.setMapperClass(BookMapper.class);
        //job设置map的输出的key类型
        job.setMapOutputKeyClass(Text.class);
        //job设置map的输出的value类型
        job.setMapOutputValueClass(IntWritable.class);

        //job设置shuffle，设置自定义了分区
        //job设置shuffle，设置局部聚合 combiner
        job.setCombinerClass(BookCombiner.class);

        //job设置reduce的实现类
        job.setReducerClass(BookReducer.class);
        //job设置reduce的输出key
        job.setOutputKeyClass(Text.class);
        //job设置reduce的输出value
        job.setOutputValueClass(IntWritable.class);

        //job设置reducer输出到文件的输出类型
        job.setOutputFormatClass(TextOutputFormat.class);
        //job设置输出的路径
        TextOutputFormat.setOutputPath(job, new Path("D:\\project\\mr\\data\\books_result"));

        //job等待执行
        boolean flag = job.waitForCompletion(true);
        //返回程序执行的结果，如果成功返回为0，否则1
        System.exit(flag ? 0 : 1);
    }
}
